/*
 * Copyright (C) 2008 CERNET Network Center 
 * dvping/dvmcast/hdvping/hdvmcast series 
 * 
 * Design and coding: 
 *  	Xing Li <xing@cernet.edu.cn> 
 *	Congxiao Bao <congxiao@cernet.edu.cn>
 * 	Jinpeng Jiang <jjp02@mails.tsinghua.edu.cn>
 * 
 * Contributions:
 * 
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 */


#include <Socket.h>
using namespace kascale;
#include <iostream>
#include <vector>
using namespace std;
#include <FileSystemUtil.h>
using namespace System::IO;
#include "dvsource.h"
#include "dvdest.h"
#include "FileDVSource.h"
#include "NetworkDVSource.h"
#include "DestSpecDVDestination.h"
#include "SelfReflectDVDestination.h"
#include "MeasureDVDestination.h"
#include "rtp.h"
#include "dvlog.h"

osyncstream dvlog;

void PrintUsage()
{
	cout << "dvmcast [-h] [-t time_limit]" << endl;
	cout << "        [-l log_file] [-lp log_period] [-ot open_th] [-ft fluc_th]" << endl;
	cout << "        [-noloop] [-dvf dv_file] " << endl;
	cout << "        [-4|-6] [-rp rcv_port] [-asm grp] [-ssm src grp] [-sf src_flt] [-sff sf_file]" << endl;
	cout << "        [-ech ech_port] [-d [dst_addr] [dst_port]] [-df dst_file] " << endl;
	return; 
}

double ParseBandwidth(string arg)
{
	istringstream istr(arg);
	double bandwidth = 0.0;
	istr >> bandwidth;
	if(!istr || bandwidth <= 0.0)
	{
		return -1.0;
	}
	string unit;
	istr >> unit;
	if(istr && !unit.empty())
	{
		tolower(unit);
        switch(unit[0])
		{
		case 't':
			bandwidth *= 1e12;
			break;
		case 'g':
			bandwidth *= 1e9;
			break;
		case 'm':
			bandwidth *= 1e6;
			break;
		case 'k':
			bandwidth *= 1e3;
			break;
		default:
			return -1.0;
		}
	}
	return bandwidth;
}


int main(int argc, char* argv[])
{
	Socket::Init();
	u_int32_t time_limit = 0;
	bool loop = true;
	bool measure = false;
	bool write_log = false;
	string log_file;
	int32_t log_period = 1;
	string dv_file;
	in_port_t rcv_port = 8000;
	IPAddress asm_grp;
	IPAddress ssm_src;
	IPAddress ssm_grp;
	vector<IPAddressFilter> src_filters;
	int src_filter_length = 0;
	vector<IPAddress> dst_addrs;
	vector<in_port_t> dst_ports;
	string dst_file;
	sa_family_t sa_family = AF_UNSPEC;
	
	double open_thresh = 0.5e6;		// 0.5Mbps
	double fluc_thresh = 0.5e6;

    for(int i = 1; i < argc; ++i)
	{
		string arg = tolower(argv[i]);
		if(arg == "-h")
		{
			PrintUsage();
			return 1;
		}
		else if(arg == "-t")
		{
			++i;
			if(i < argc)
			{
				if(!ParseString(argv[i], time_limit))
				{
					cout << "Invalid time_limit:" << argv[i] << endl;
					PrintUsage();
					return 1;
				}
			}
			else
			{
				cout << "no time limit" << endl;
				PrintUsage();
				return 1;
			}
		}
		else if(arg == "-l")
		{
			if(arg.find('l') != string::npos)
			{
				write_log = true;
			}
			++i;
			if(i < argc)
			{
				if(argv[i][0] == '-')
				{
					--i;
					log_file = "";
				}
				else
				{
					log_file = argv[i];
				}
			}
			else
			{
				--i;
				log_file = "";
			}
		}
		else if(arg == "-lp")
		{
			++i;
			if(i < argc)
			{
				if(!ParseString(argv[i], log_period))
				{
					cout << "Invalid log period:" << argv[i] << endl;
					PrintUsage();
					return 1;
				}
				if(log_period < 1)
				{
					cout << "Log period out of range:" << argv[i] << endl;
					PrintUsage();
					return 1;
				}
			}
			else
			{
				cout << "no log_period" << endl;
				PrintUsage();
				return 1;
			}
		}
		else if(arg == "-ot")
		{
			++i;
			if(i < argc)
			{
				open_thresh = ParseBandwidth(argv[i]);
				if(open_thresh <= 0.0)
				{
					cout << "Invalid open threshold." << argv[i] << endl;
					PrintUsage();
					return 1;
				}
			}
			else
			{
				cout << "no open threshold." << endl;
				PrintUsage();
				return 1;
			}
		}
		else if(arg == "-ft")
		{
			++i;
			if(i < argc)
			{
				fluc_thresh = ParseBandwidth(argv[i]);
				if(fluc_thresh <= 0.0)
				{
					cout << "Invalid fluctuate threshold." << argv[i] << endl;
					PrintUsage();
					return 1;
				}
			}
			else
			{
				cout << "no fluctuate threshold." << endl;
				PrintUsage();
				return 1;
			}
		}
		else if(arg == "-noloop")
		{
			loop = false;
		}
		else if(arg == "-dvf")
		{
			++i;
			if(i < argc)
			{
				dv_file = argv[i];
				if(!File::Exists(dv_file))
				{
					cout << "file " << dv_file << " doesn't exist." << endl;
					PrintUsage();
					return 1;
				}
			}
			else
			{
				cout << "no dv_file" << endl;
				PrintUsage();
				return 1;
			}
		}
		else if(arg == "-4")
		{
			if(sa_family == AF_INET6)
			{
				cout << "Cannot specify both \"-4\" and \"-6\" option" << endl;
				PrintUsage();
				return 1;
			}
			sa_family = AF_INET;
		}
		else if(arg == "-6")
		{
			if(sa_family == AF_INET)
			{
				cout << "Cannot specify both \"-4\" and \"-6\" option" << endl;
				PrintUsage();
				return 1;
			}
			sa_family = AF_INET6;
		}
		else if(arg == "-rp")
		{
			++i;
			if(i < argc)
			{
				if(!ParseString(argv[i], rcv_port))
				{
					cout << "Invalid receive port:" << argv[i] << endl;
					PrintUsage();
					return 1;
				}
			}
			else
			{
				cout << "no rcv_port" << endl;
				PrintUsage();
				return 1;
			}
		}
		else if(arg == "-asm")
		{
			++i;
			if(i < argc)
			{
				if(!ParseString(argv[i], asm_grp) || !asm_grp.IsMulticast())
				{
					cout << "Invalid asm multicast address:" << argv[i] << endl;
					PrintUsage();
					return 1;
				}
			}
			else
			{
				cout << "no asm_grp" << endl;
				PrintUsage();
				return 1;
			}
		}
		else if(arg == "-ssm")
		{
			++i;
			if(i < argc - 1)
			{
				if(!ParseString(argv[i], ssm_src) || !ssm_src.IsValid())
				{
					cout << "Invalid ssm source address:" << argv[i] << endl;
					PrintUsage();
					return 1;
				}
				++i;
				if(!ParseString(argv[i], ssm_grp) || !ssm_grp.IsMulticast())
				{
					cout << "Invalid ssm group address:" << argv[i] << endl;
					PrintUsage();
					return 1;
				}
			}
			else
			{
				cout << "no ssm_src and ssm_grp" << endl;
				PrintUsage();
				return 1;
			}
		}
		else if(arg == "-sf")
		{
			++i;
			if(i < argc)
			{
				IPAddressFilter flt;
				if(!ParseString(argv[i], flt))
				{
					cout << "Invalid src_filter:" << argv[i] << endl;
					PrintUsage();
					return 1;
				}
				src_filters.push_back(flt);
			}
			else
			{
				cout << "no src_filter" << endl;
				PrintUsage();
				return 1;
			}
		}
		else if(arg == "-sff")
		{
			++i;
			if(i < argc)
			{
				if(!File::Exists(argv[i]))
				{
					cout << "source filter file:" << argv[i] << " doesn't exist." << endl;
					PrintUsage();
					return 1;
				}
				ifstream fin(argv[i]);
				IPAddressFilter flt;
				fin >> flt;
				while(fin)
				{
					src_filters.push_back(flt);
					fin >> flt;
				}
				fin.close();
			}
			else
			{
				cout << "no sf_file" << endl;
				PrintUsage();
				return 1;
			}
		}
		else if(arg == "-d")
		{
			++i;
			IPAddress dst_addr;
			bool dst_addr_set = false;
			in_port_t dst_port = 8000;
			bool dst_port_set =false;
			if(i < argc)
			{
				arg = argv[i];
				if(arg.find_first_of(".:") != string::npos)
				{
					dst_addr_set = ParseString(argv[i], dst_addr);
				}
				else
				{
					dst_port_set = ParseString(argv[i], dst_port);
				}
				if(!dst_addr_set && !dst_port_set)
				{
					dst_addr_set = true;
					dst_port_set = true;
					dst_port = 8000;
					dst_addr = IPAddress::Null;
				}
				if(i < argc - 1)
				{
					if(!dst_addr_set)
					{
						++i;
						if(!ParseString(argv[i], dst_addr))
						{
							--i;
						}
						else
						{
							dst_addr_set = true;
						}
					}
					if(!dst_port_set)
					{
						++i;
						if(!ParseString(argv[i], dst_port))
						{
							--i;
						}
						else
						{
							dst_port_set= true;
						}
					}
				}
			}
			else
			{
				--i;
			}
			if(!dst_addr_set)
			{
				dst_addr = IPAddress::Null;			// null address means reflect data back to the sender
			}
			if(!dst_port_set)
			{
				dst_port = 8000;
			}
			dst_addrs.push_back(dst_addr);
			dst_ports.push_back(dst_port);
		}
		else if(arg == "-ech")
		{
			++i;
			in_port_t ech_port = 8000;
			if(i < argc)
			{
				if(!ParseString(argv[i], ech_port))
				{
					--i;
					ech_port = 8000;
				}
			}
			else
			{
				--i;
			}
			dst_addrs.push_back(IPAddress::Null);
			dst_ports.push_back(ech_port);
		}
		else if(arg == "-df")
		{
			++i;
			if(i < argc)
			{
				ifstream fin(argv[i]);
				if(!fin)
				{
					cout << "dest file:" << argv[i] << " cannot open or doesn't exist." << endl;
					PrintUsage();
					return 1;
				}
				IPAddress dst_addr;
				in_port_t dst_port;
				fin >> dst_addr;
				fin >> dst_port;
				while(fin)
				{
					dst_addrs.push_back(dst_addr);
					dst_ports.push_back(dst_port);
					fin >> dst_addr;
					fin >> dst_port;
				}
				fin.close();
			}
			else
			{
				cout << "no dst_file" << endl;
				PrintUsage();
				return 1;
			}
		}
	}
	if(measure || write_log)
	{
		dvlog.open(log_file);
	}
	if(sa_family == AF_UNSPEC)
	{
		sa_family = AF_INET;
	}
	IDVSource* pSource = NULL;
	// file source
	if(!dv_file.empty())
	{
		pSource = new FileDVSource(dv_file, loop);
		if(pSource == NULL)
		{
			cout << "Create file dv source failed." << endl;
			return 1;
		}
	}
	else		//network dv source
	{
		NetworkDVSource* pNetSrc = new NetworkDVSource(sa_family, rcv_port, src_filters, write_log ? log_period : 0);
		if(asm_grp.IsMulticast())
		{
			if(asm_grp.AddressFamily() != sa_family)
			{
				cout << "asm_grp's address family doesn't match." << endl;
				return 1;
			}
			if(!pNetSrc->JoinGroup(asm_grp))
			{
				cout << "Join group :" << asm_grp << " failed." << endl;
				return 1;
			}
		}
		else if(ssm_src.IsValid() && ssm_grp.IsMulticast())
		{
			if(ssm_src.AddressFamily() != sa_family)
			{
				cout << "ssm_src's address family doesn't match." << endl;
				return 1;
			}
			if(ssm_grp.AddressFamily() != sa_family)
			{
				cout << "ssm_grp's address family doesn't match." << endl;
				return 1;
			}
			if(!pNetSrc->JoinSourceGroup(ssm_src, ssm_grp))
			{
				cout << "Join source group:(" << ssm_src << ", " << ssm_grp << ")" << endl;
				return 1;
			}
		}
		pNetSrc->SetInitLogThreshold(open_thresh / 8e6);
		pNetSrc->SetFluctuateThreshold(fluc_thresh / 8e6);
		pSource = pNetSrc;
	}
    for(u_int32_t i = 0; i < dst_addrs.size(); ++i)
	{
		IDVDestination * pDest = NULL;
		if(dst_addrs[i] == IPAddress::Null)
		{
			pDest = new SelfReflectDVDestination(sa_family, dst_ports[i], write_log ? log_period : 0);
		}
		else
		{
			pDest = new DestSpecDVDestination(dst_addrs[i], dst_ports[i], write_log ? log_period : 0);
		}
		pSource->AddDestination(pDest);
	}
	/*
	if(measure)
	{
		pSource->AddDestination(new MeasureDVDestination(log_period));
	}
	*/
	pSource->Start();
	int64_t start = DateTime::Now().GetTickCount();
	if(time_limit == 0)
	{
        while(pSource->IsRunning())
		{
			Thread::Sleep(100000LL);
		}
	}
	else
	{
		int64_t start_time = DateTime::Now().GetTickCount();
		while(pSource->IsRunning())
		{
			Thread::Sleep(100000LL);
			int64_t now = DateTime::Now().GetTickCount();
			if(now - start_time >= time_limit * 1000000LL)
			{
				break;
			}
		}
	}
	pSource->Stop();
	delete pSource;
	dvlog.close();
	Socket::Cleanup();
	return 0;
}

